package com.huitongjy.common.zokeeper;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.io.InputStream;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.text.DecimalFormat;
import java.util.Comparator;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;

/**
 * 自动清理废弃节点的
 * @author guan
 */
public class AutoFlushZooKeeperWorkId {
    private static final String PATH_SYSTEM = "/www/worker/system";
    private static final String PATH_GLOBAL = "/www/worker/global";


    private ZooKeeper zookeeper;

    /**
     * 项目内私有的workerId
     */
    private Long workerId;
    /**
     * 项目间通用的workerId
     */
    private Long globalWorkerId;

    /**
     * zookeeper地址
     */
    private String address;

    /**
     * 项目
     */
    private String nodeName;

    private AutoFlushZooKeeperWorkId(String address, String nodeName) {
        this.address = address;
        this.nodeName = nodeName;
        this.init();
    }

    private static AutoFlushZooKeeperWorkId INSTANCE;

    public static synchronized AutoFlushZooKeeperWorkId getInstanse() {
        if(INSTANCE==null) {
            Properties properties = new Properties();
            try {
                //项目使用dubbo的zookeeer配置
                InputStream resourceAsStream = AutoFlushZooKeeperWorkId.class.getClassLoader().getResourceAsStream("dubbo.properties");
                if(resourceAsStream==null) {
                    throw new RuntimeException("主键生成配置不存在");
                }

                properties.load(resourceAsStream);
                String address = (String) properties.get("dubbo.registry.address");
                String nodeName = (String) properties.get("dubbo.application.name");
                if(address==null || nodeName ==null) {
                    throw new RuntimeException("主键生成配置不存在");
                }
                INSTANCE = new AutoFlushZooKeeperWorkId(address, nodeName);
            } catch (IOException ignore) {
                throw new RuntimeException("主键生成器配置不存在");
            }
        }
        return INSTANCE;
    }

    private void init() {
        try {
            CountDownLatch countDownLatch = new CountDownLatch(1);
            this.zookeeper = new ZooKeeper(address, 20000, event -> {
                if(event.getState() == Watcher.Event.KeeperState.SyncConnected){
                    countDownLatch.countDown();
                }
            });
            countDownLatch.await();
            System.out.println("zookeeper connect ok");
        } catch (Exception e) {
            System.out.println("zookeeper connect error");
            throw new RuntimeException(e);
        }
    }



    public long getWorkerId() {
        if(workerId==null) {
            try {
                // init workerId
                workerId = getZookeeperWorkerId(false);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        return workerId;
    }

    public long getGlobalWorkerId() {
        if(globalWorkerId==null) {
            try {
                // init global workerId
                globalWorkerId = getZookeeperWorkerId(true);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        return globalWorkerId;
    }


    /**
     * 获取一个自增的机器Id (项目|host)唯一, 如果没有则生成
     * @param isGlobal 全局
     * @return
     * @throws KeeperException
     * @throws InterruptedException
     */
    private Long getZookeeperWorkerId(boolean isGlobal) {
        String nodePath = getNodePath(isGlobal);
        String workerIdPath = getWorkerIdPath(isGlobal);
        String node = nodePath + "/" + getLocalIP();

        try {
            mkdir(nodePath);

            String workerId;
            String workIdNode;
            Stat exists = zookeeper.exists(node, false);

            if(exists==null) {
                //本机器 从来没有使用过
                workerId = initNewWorkerId(workerIdPath);
//                if(workerId ==null) {
//                    int i = flushWorkerId(nodePath, workerIdPath);
//                    if(i>0) {
//                        workerId = initNewWorkerId(workerIdPath);
//                    }
//                }

                if(workerId ==null) {
                    throw new RuntimeException("worker节点已满,请手动清除不用节点");
                }

                zookeeper.create(node, workerId.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                System.out.println(node + "节点创建成功！");
                workIdNode = workerIdPath + "/" + workerId;
            } else {
                //本机器 已经使用过
                byte[] data = zookeeper.getData(node, false, exists);
                workerId = new String(data);
                workIdNode = workerIdPath + "/" + workerId;

                byte[] lastTime = zookeeper.getData(workIdNode, false, new Stat());
                if(Long.valueOf(new String(lastTime)) + 3000L > System.currentTimeMillis()) {
                    throw new RuntimeException("节点时间回调,会生成重复主键");
                }
            }
//            new WorkerIdThread(workIdNode).start();
            return transLong(workerId);
        } catch (KeeperException | InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 生成一个新的workerId
     * @param workerIdPath
     * @return
     * @throws KeeperException
     * @throws InterruptedException
     */
    private String initNewWorkerId(String workerIdPath) throws KeeperException, InterruptedException {
        mkdir(workerIdPath);

        String workerId = null;

        List<String> children = zookeeper.getChildren(workerIdPath, false);
        children.sort(Comparator.naturalOrder());
        for (int i = 0; i < Integer.MAX_VALUE; i++) {
            if(i == children.size() ||  i != Integer.valueOf(children.get(i))) {
                workerId = transStr(i);
                break;
            }
        }
        if(workerId!=null) {
            String result = zookeeper.create(workerIdPath + "/" + workerId, getCurrTimeMillis().getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println(result + "节点创建成功！");
        }
        return workerId;
    }

    /**
     * 清除不用的节点
     * @return
     * @param nodePath
     * @param workIdPath
     */
    private int flushWorkerId(String nodePath, String workIdPath) throws KeeperException, InterruptedException {
        List<String> children = zookeeper.getChildren(nodePath, false);

        int i=0;
        for (String child : children) {
            try {
                byte[] nodeData = zookeeper.getData(nodePath + "/" + child, false, new Stat());
                String workerId = new String(nodeData);
                byte[] workerData = zookeeper.getData(workIdPath + "/" + workerId, false, new Stat());
                //超过5天没有更新,认为数据不再使用
                if(System.currentTimeMillis() - Long.valueOf(new String(workerData)) > 1000 * 60 * 60 * 24 * 5) {
                    zookeeper.delete(workIdPath + "/" + workerId, -1);
                    zookeeper.delete(nodePath + "/" + child, -1);
                    i++;
                }
            } catch (KeeperException | InterruptedException | NumberFormatException e) {
                e.printStackTrace();
            }
        }
        return i;
    }

    /**
     * 创建目录
     * @param nodePath
     * @throws KeeperException
     * @throws InterruptedException
     */
    private void mkdir(String nodePath) throws KeeperException, InterruptedException {
        Stat exists = zookeeper.exists(nodePath, false);
        if(exists != null) {
            return;
        }

        int index = nodePath.lastIndexOf("/");
        if(index>0) {
            mkdir(nodePath.substring(0, index));
        }
        zookeeper.create(nodePath,null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println(nodePath + "节点创建成功！");
    }

    /**
     * 获取当前时间
     * @return
     */
    private String getCurrTimeMillis() {
        return String.valueOf(System.currentTimeMillis());
    }

    /**
     * 本机的IP地址
     * @return
     */
    private static String getLocalIP(){
        InetAddress addr;
        try {
            addr = InetAddress.getLocalHost();
        } catch (UnknownHostException e) {
            throw new RuntimeException(e);
        }

        byte[] ipAddr = addr.getAddress();
        StringBuilder ipAddrStr = new StringBuilder();
        for (int i = 0; i < ipAddr.length; i++) {
            if (i > 0) {
                ipAddrStr.append(".");
            }
            ipAddrStr.append(ipAddr[i] & 0xFF);
        }
        return ipAddrStr.toString();
    }

    private static String transStr(long value) {
        return new DecimalFormat("0000000000").format(value);
    }

    private static Long transLong(String value) {
        return Long.valueOf(value);
    }

    /**
     * 获取workId所在目录
     * @param isGolbal
     * @return
     */
    private String getWorkerIdPath(boolean isGolbal) {
        if(isGolbal) {
            return PATH_GLOBAL + "/" + "workerId";
        }
        return PATH_SYSTEM + "/" + nodeName + "/" + "workerId";
    }

    /**
     * 获取节点所在目录
     * @param isGolbal
     * @return
     */
    private String getNodePath(boolean isGolbal) {
        if(isGolbal) {
            return PATH_GLOBAL + "/" + nodeName;
        }
        return PATH_SYSTEM + "/" + nodeName;
    }

    /**
     * 异步循环更新节点时间,为了自动清除不用节点,及上次workerId最后使用时间(防止时间回拨)
     */
    class WorkerIdThread extends Thread {
        private String workIdNode;
        private int version = -1;
        WorkerIdThread(String workIdNode) {
            this.workIdNode = workIdNode;
            setDaemon(true);
        }

        @Override
        public void run() {
            while (true) {
                try {
                    //每三秒更新一次 节点时间
                    version = zookeeper.setData(workIdNode, getCurrTimeMillis().getBytes(), version).getVersion();
                    Thread.sleep(3000L);
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
